package com.data.permission.transf;

import com.alibaba.fastjson.JSONObject;
import com.data.permission.constrant.CommonConstrant;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @包名称 com.thrift.distributed.process
 * @类描述 <pre></pre>
 * @作者 zhangbo
 * @创建时间 2019/9/19
 * @版本 1.00
 * @修改记录 <pre>
 *     版本           修改人         修改日期         修改内容描述
 *     -----------------------------------------------------------
 *     1.00           Lenovo        2019/9/19
 *     -----------------------------------------------------------
 * </pre>
 */
@Slf4j
@Component
public class ClientListener implements Watcher {

    @Value("${REGIST.ZOOKEEPER:localhost:2181}")
    private String zookeeperAddress;

    private ZooKeeper zooKeeper;

    /*注册中心超时或重复检测休眠时间*/
    @Value("${ZOOKEEPER.SERVICE.SLEEP.TIME:10000}")
    private int zookeeperServiceSleepTime;

    private static Map<String, ServiceInfo> serviceInfoMap = new ConcurrentHashMap<>(8);

    public void init() {
        log.info("开始初始化数据权限客户端....");
        if (null == zooKeeper) {
            while (true) {
                try {
                    zooKeeper = new ZooKeeper(zookeeperAddress, zookeeperServiceSleepTime, this);
                    if (zooKeeper.exists(CommonConstrant.PROCESS_SERVICE_PATH, this) == null) {
                        zooKeeper.create(CommonConstrant.PROCESS_SERVICE_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                    }
                    updateServiceMap();
                    ClientRun.COUNT_DOWN_LATCH.countDown();
                    break;
                } catch (IOException e) {
                    log.error("创建连接失败{}", e);
                } catch (InterruptedException e) {
                    log.error("线程中断");
                    Thread.currentThread().interrupt();
                } catch (KeeperException e) {
                    log.error("创建zookeeper节点异常");
                    try {
                        if (null != zooKeeper) {
                            zooKeeper.close();
                        }
                    } catch (InterruptedException e1) {
                        log.error("线程休眠中断：{}", e1);
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }
        log.info("数据权限客户端初始化完成....");
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        log.debug("客户端监听到事件：{}", watchedEvent);
        if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
            updateServiceMap();
        }
    }

    /**
     * 更新服务端信息
     */
    public void updateServiceMap() {
        log.debug("刷新服务列表");
        try {
            if(null == zooKeeper) {
                zooKeeper = new ZooKeeper(zookeeperAddress, zookeeperServiceSleepTime, this);
                zooKeeper.register(this);
            }
            List<String> serviceList = zooKeeper.getChildren(CommonConstrant.PROCESS_SERVICE_PATH, this);
            if (null != serviceList) {
                for (String service : serviceList) {
                    String[] serviceInfo = service.split(":");
                    if (serviceInfo.length == 2) {
                        ServiceInfo serverInfo = new ServiceInfo(serviceInfo[0], serviceInfo[1]);
                        serviceInfoMap.put(service, serverInfo);
                    }
                }
                Set<String> serviceKeys = serviceInfoMap.keySet();
                Iterator<String> iterator = serviceKeys.iterator();
                while (iterator.hasNext()) {
                    String serviceKey = iterator.next();
                        /*如果zk中服务已经不存在了，则需要移除掉*/
                    if (!serviceList.contains(serviceKey)) {
                        serviceInfoMap.remove(serviceKey);
                    }
                }
            }
            log.info("当前服务列表：{}", JSONObject.toJSONString(serviceInfoMap));
        } catch (KeeperException e) {
            log.error("创建监听节点异常：{}", e);
            try {
                if(null != zooKeeper) {
                    zooKeeper.close();
                }
                zooKeeper = new ZooKeeper(zookeeperAddress, zookeeperServiceSleepTime, this);
                zooKeeper.register(this);
                updateServiceMap();
            } catch (InterruptedException e1) {
                log.error("线程中断：{}", e1);
                Thread.currentThread().interrupt();
            } catch (IOException e1) {
                log.error("创建zookeeper异常：{}", e);
            }
        } catch (InterruptedException e) {
            log.error("线程中断：{}", e);
            Thread.currentThread().interrupt();
        } catch (IOException e) {
            log.error("创建zookeeper异常：{}", e);
        }
    }

    public static Map<String, ServiceInfo> getServiceInfoMap() {
        return serviceInfoMap;
    }

    /**
     * 随机获取服务列表
     *
     * @return
     */
    public static ServiceInfo roundServiceInfo() {
        if (serviceInfoMap.size() == 0) {
            log.error("当前无注册服务可用");
            return null;
        }
        int serverSize = serviceInfoMap.size();
        Object[] keyList = serviceInfoMap.keySet().toArray();
        if (serverSize > 1) {
            return serviceInfoMap.get(keyList[new Random().nextInt(serverSize)]);
        }
        return serviceInfoMap.get(keyList[0]);
    }

    @PreDestroy
    public void close() {
        if (null != zooKeeper) {
            try {
                zooKeeper.close();
            } catch (InterruptedException e) {
                log.error("关闭zookeeper连接错误：{}", e);
                Thread.currentThread().interrupt();
            }
        }
    }
}
